package com.sunday.common.mq.kafka.study.spring.e6_Receiving_Messages_1;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.ContainerCustomizer;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;

import java.nio.ByteBuffer;

/**
 * https://docs.spring.io/spring-kafka/docs/current/reference/html/#receiving-messages
 * {@link org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration}
 * <p>
 * 消息侦听器
 * 使用消息侦听器容器时，必须提供侦听器来接收数据。 消息侦听器目前支持八种接口。 以下列表显示了这些接口：
 * <p>
 * 使用此接口处理在使用自动提交或容器管理的提交方法之一时从 Kafka 使用者操作接收的单个实例。ConsumerRecord poll()
 * public interface MessageListener<K, V> {
 * <p>
 * void onMessage(ConsumerRecord<K, V> data);
 * <p>
 * }
 * <p>
 * 使用此接口用于处理在使用其中一种手动提交方法时从 Kafka 使用者操作接收的单个实例。ConsumerRecord poll()
 * public interface AcknowledgingMessageListener<K, V> {
 * <p>
 * void onMessage(ConsumerRecord<K, V> data, Acknowledgment acknowledgment);
 * <p>
 * }
 * <p>
 * 使用此接口处理在使用自动提交或容器管理的提交方法之一时从 Kafka 使用者操作接收的单个实例。 提供对对象的访问。ConsumerRecord poll() Consumer
 * public interface ConsumerAwareMessageListener<K, V> extends MessageListener<K, V> {
 * <p>
 * void onMessage(ConsumerRecord<K, V> data, Consumer<?, ?> consumer);
 * <p>
 * }
 * <p>
 * 使用此接口用于处理在使用其中一种手动提交方法时从 Kafka 使用者操作接收的单个实例。 提供对对象的访问。ConsumerRecord poll() Consumer
 * public interface AcknowledgingConsumerAwareMessageListener<K, V> extends MessageListener<K, V> {
 * <p>
 * void onMessage(ConsumerRecord<K, V> data, Acknowledgment acknowledgment, Consumer<?, ?> consumer);
 * <p>
 * }
 * <p>
 * 使用此接口在使用自动提交或容器管理的提交方法之一时处理从 Kafka 使用者操作接收的所有实例。 使用此接口时不受支持，因为侦听器将获得完整的批处理。ConsumerRecord poll() AckMode.RECORD
 * public interface BatchMessageListener<K, V> {
 * <p>
 * void onMessage(List<ConsumerRecord<K, V>> data);
 * <p>
 * }
 * <p>
 * 使用此接口处理在使用其中一种手动提交方法时从 Kafka 使用者操作接收的所有实例。ConsumerRecord poll()
 * public interface BatchAcknowledgingMessageListener<K, V> {
 * <p>
 * void onMessage(List<ConsumerRecord<K, V>> data, Acknowledgment acknowledgment);
 * <p>
 * }
 * <p>
 * 使用此接口在使用自动提交或容器管理的提交方法之一时处理从 Kafka 使用者操作接收的所有实例。 使用此接口时不受支持，因为侦听器将获得完整的批处理。 提供对对象的访问。ConsumerRecord poll() AckMode.RECORDConsumer
 * public interface BatchConsumerAwareMessageListener<K, V> extends BatchMessageListener<K, V> {
 * <p>
 * void onMessage(List<ConsumerRecord<K, V>> data, Consumer<?, ?> consumer);
 * <p>
 * }
 * <p>
 * 使用此接口处理在使用其中一种手动提交方法时从 Kafka 使用者操作接收的所有实例。 提供对对象的访问。ConsumerRecord poll() Consumer
 * public interface BatchAcknowledgingConsumerAwareMessageListener<K, V> extends BatchMessageListener<K, V> {
 * <p>
 * void onMessage(List<ConsumerRecord<K, V>> data, Acknowledgment acknowledgment, Consumer<?, ?> consumer);
 * <p>
 * }
 */
@Slf4j
@SpringBootApplication
public class Application {

    /**
     * {@link org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration}
     * {@link org.springframework.boot.autoconfigure.kafka.KafkaAnnotationDrivenConfiguration.EnableKafkaConfiguration}
     * {@link org.springframework.kafka.annotation.EnableKafka}
     * {@link org.springframework.kafka.annotation.KafkaListenerConfigurationSelector}
     * {@link org.springframework.kafka.annotation.KafkaBootstrapConfiguration}
     * {@link org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor#postProcessAfterInitialization(Object, String)}
     * {@link org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor#processListener}
     * {@link org.springframework.kafka.config.KafkaListenerEndpointRegistrar#registerEndpoint(org.springframework.kafka.config.KafkaListenerEndpoint, org.springframework.kafka.config.KafkaListenerContainerFactory<?>)}
     * {@link org.springframework.kafka.listener.KafkaMessageListenerContainer.ListenerConsumer#run}
     * {@link org.springframework.kafka.config.KafkaListenerEndpointRegistry#start}
     */

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

    /**
     * https://docs.spring.io/spring-kafka/docs/current/reference/html/#using-ConcurrentMessageListenerContainer
     * 负载策略
     * {@link org.apache.kafka.clients.consumer.internals.ConsumerCoordinator#onJoinComplete(int, String, String, ByteBuffer)}
     */
    /**
     * https://docs.spring.io/spring-kafka/docs/current/reference/html/#record-listener
     */
    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Object, Object>> kafkaListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ObjectProvider<ConsumerFactory<Object, Object>> kafkaConsumerFactory,
            ObjectProvider<ContainerCustomizer<Object, Object, ConcurrentMessageListenerContainer<Object, Object>>> kafkaContainerCustomizer,
            KafkaProperties properties
    ) {
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        configurer.configure(factory, kafkaConsumerFactory
                .getIfAvailable(() -> new DefaultKafkaConsumerFactory<>(properties.buildConsumerProperties())));
        kafkaContainerCustomizer.ifAvailable(factory::setContainerCustomizer);
        return factory;
    }

    /**
     * https://docs.spring.io/spring-kafka/docs/current/reference/html/#kafkalistener-attrs
     *
     * @KafkaListener属性修改 从版本 2.7.2 开始，您现在可以在创建容器之前以编程方式修改注释属性。 为此，请将一个或多个添加到应用程序上下文中。
     * 是 并且必须返回属性映射。 属性值可以包含 SpEL 和/或属性占位符;在执行任何分辨率之前调用增强器。 如果存在多个增强器，
     * 并且它们实现了 ，则将按顺序调用它们。
     * {@link org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor#buildEnhancer()}
     */
    @Bean
    public static KafkaListenerAnnotationBeanPostProcessor.AnnotationEnhancer groupIdEnhancer() {
        return (attrs, element) -> {
//            attrs.put("groupId", attrs.get("id") + "." + (element instanceof Class
//                    ? ((Class<?>) element).getSimpleName()
//                    : ((Method) element).getDeclaringClass().getSimpleName()
//                    +  "." + ((Method) element).getName()));
            attrs.put("groupId", attrs.get("id"));
            return attrs;
        };
    }

    @Bean
    public NewTopic topic1() {
        return TopicBuilder.name("topic1")
                .partitions(3)
                .replicas(1)
                .build();
    }

    @Bean(name = "myListener")
    public Listener myListener() {
        return new Listener("topic1");
    }

}
